#include "mqtt.h"

#define QOS 1
#define TIMEOUT 10000L
static MQTTClient mqtt_client;

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    // xx_log_debug("topic: %s", topicName);
    // xx_log_debug("message: %.*s", message->payloadlen, (char *)message->payload);
    uv_async_t *handle = (uv_async_t *)context;
    char* buf = (char *)malloc(message->payloadlen);
    memcpy(buf, message->payload, message->payloadlen);
    uv_handle_set_data((uv_handle_t*)handle, (void *)buf);
    uv_async_send(handle);
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void conn_lost(void *context, char *cause)
{
    xx_log_warn("mqtt connection lost, cause: %s", cause);
}

int mqtt_client_init(sys_ctrl_t *ctrl)
{
    int rc;
    rc = MQTTClient_create(&mqtt_client, ctrl->cfg.mqtt.addr, ctrl->cfg.mqtt.whois,
                           MQTTCLIENT_PERSISTENCE_NONE, NULL);
    if (rc)
    {
        xx_log_error("Failed to create mqtt_client, return code %d\n", rc);
        return 1;
    }

    rc = MQTTClient_setCallbacks(mqtt_client, (void*)&ctrl->async_mqtt, conn_lost, msgarrvd, NULL);
    if (rc)
    {
        xx_log_error("Failed to set callbacks, return code %d\n", rc);
        MQTTClient_destroy(&mqtt_client);
        return 2;
    }

    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    // conn_opts.username = UserName;          // 用户名
    // conn_opts.password = PassWord;          // 用户名对应的密码
    conn_opts.keepAliveInterval = 60;
    conn_opts.cleansession = 1;
    rc = MQTTClient_connect(mqtt_client, &conn_opts);
    if (rc)
    {
        printf("Failed to connect, return code %d\n", rc);
        MQTTClient_destroy(&mqtt_client);
        return 3;
    }

    for (size_t i = 0; i < ctrl->cfg.mqtt.sub_count; i++)
    {
        xx_log_info("Subscribing to topic %s\nfor mqtt_client %s using QoS%d", ctrl->cfg.mqtt.sub[i], ctrl->cfg.mqtt.whois, QOS);
        rc = MQTTClient_subscribe(mqtt_client, ctrl->cfg.mqtt.sub[i], QOS);
        if (rc)
        {
            printf("Failed to subscribe, return code %d\n", rc);
            rc = 4;
            mqtt_client_destory(ctrl);
        }
    }

    return rc;
}

int mqtt_client_destory(sys_ctrl_t *ctrl)
{
    int rc = 0;
    for (size_t i = 0; i < ctrl->cfg.mqtt.sub_count; i++)
    {
        rc = MQTTClient_unsubscribe(mqtt_client, ctrl->cfg.mqtt.sub[i]);
        if (rc)
            xx_log_warn("Failed to unsubscribe, return code %d", rc);
    }

    rc = MQTTClient_disconnect(mqtt_client, TIMEOUT);
    if (rc)
        xx_log_info("Failed to disconnect, return code %d", rc);

    MQTTClient_destroy(&mqtt_client);
    return rc;
}

int mqtt_client_publish(sys_ctrl_t *ctrl, const char *data, int len)
{
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    pubmsg.payload = (void *)data;
    pubmsg.payloadlen = len;
    pubmsg.qos = 1;      // qos等级为1
    pubmsg.retained = 0; // 服务器不保留消息
    if (MQTTClient_publishMessage(mqtt_client, ctrl->cfg.mqtt.whois, &pubmsg, NULL))
    {
        xx_log_error("Failed to publish message");
        mqtt_client_destory(ctrl);
        return 1;
    }
    return 0;
}